package com.bw.gmall.realtime.common.util;

import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.bean.TableProcessDim;
import com.bw.gmall.realtime.common.constant.Constant;
import com.mysql.cj.protocol.ResultsetRowsOwner;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;

public class DimsinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcessDim>> {

    private Connection hBaseConnection;

    @Override
    public void open(Configuration parameters) throws Exception {
        hBaseConnection = HBaseUtil.getHBaseConnection();

    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeHBaseConn(hBaseConnection);
    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcessDim> value, Context context) throws Exception {
        //主流
        JSONObject f0 = value.f0;
        //配置表
        TableProcessDim tableProcessDim = value.f1;

        System.out.println("正在写入维度数据"+tableProcessDim.getSourceTable());

        //获取主流中的类型type
        JSONObject data = f0.getJSONObject("data");

        String type = f0.getString("type");

        //要存入hbase里的表名
        String sinkTable = tableProcessDim.getSinkTable();
        String sinkRowKey = tableProcessDim.getSinkRowKey();
        String sinkFamily = tableProcessDim.getSinkFamily();


        //拿到rowkey 具体的值
        String rowkeyvalue = data.getString(sinkRowKey);


        if("delete".equals(type)){
            HBaseUtil.deleteCells(hBaseConnection, Constant.HBASE_NAMESPACE,sinkTable,rowkeyvalue);
        }else {
            HBaseUtil.putCells(hBaseConnection,Constant.HBASE_NAMESPACE,sinkTable,rowkeyvalue,sinkFamily,data);
        }


    }
}
